--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit fa98acbab0a03b6b1df21c475d405cffb86dae1b
Parents : 3e5fab8
Author : Sudo-Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-03-05T22:12:30-06:00
Refactor CommunityInterfacesManager to remove health check functionality and simplify interface retrieval. Update tests to reflect static interface list behavior.
Changes
2 files changed, 25 insertions(+), 101 deletions(-)
Diff
diff --git a/meshchatx/src/backend/community_interfaces.py b/meshchatx/src/backend/community_interfaces.py
index 914e6752..448c0c49 100644
--- a/meshchatx/src/backend/community_interfaces.py
+++ b/meshchatx/src/backend/community_interfaces.py
@@ -1,82 +1,25 @@
-import asyncio
-import time
from typing import Any
class CommunityInterfacesManager:
+ """
+ Suggested community interfaces for the interface wizard.
+ No outbound connectivity checks are performed; listing these does not contact the internet.
+ """
+
def __init__(self):
self.interfaces = [
{
- "name": "RNS Testnet Amsterdam",
- "type": "TCPClientInterface",
- "target_host": "amsterdam.connect.reticulum.network",
- "target_port": 4965,
- "description": "Reticulum Testnet Hub",
- },
- {
- "name": "RNS Testnet BetweenTheBorders",
+ "name": "Quad4",
"type": "TCPClientInterface",
- "target_host": "reticulum.betweentheborders.com",
- "target_port": 4242,
- "description": "Reticulum Testnet Hub",
+ "target_host": "62.151.179.77",
+ "target_port": 45657,
+ "description": "Quad4 Official Node",
},
]
- self.status_cache = {}
- self.last_check = 0
- self.check_interval = 600 # Check every 10 minutes
-
- async def check_health(self, host: str, port: int) -> bool:
- try:
- # Simple TCP connect check as a proxy for "working"
- # In a real RNS environment, we might want to use RNS.Transport.probe()
- # but that requires Reticulum to be running with a configured interface to that target.
- # For "suggested" interfaces, we just check if they are reachable.
- reader, writer = await asyncio.wait_for(
- asyncio.open_connection(host, port),
- timeout=3.0,
- )
- writer.close()
- await writer.wait_closed()
- return True
- except Exception:
- return False
- async def update_statuses(self):
- tasks = [
- self.check_health(iface["target_host"], iface["target_port"])
+ async def get_interfaces(self) -> list[dict[str, Any]]:
+ return [
+ {**iface, "online": None, "last_check": 0}
for iface in self.interfaces
]
-
- results = await asyncio.gather(*tasks)
-
- for iface, is_online in zip(self.interfaces, results):
- self.status_cache[iface["name"]] = {
- "online": is_online,
- "last_check": time.time(),
- }
- self.last_check = time.time()
-
- async def get_interfaces(self) -> list[dict[str, Any]]:
- # If cache is old or empty, update it
- if time.time() - self.last_check > self.check_interval or not self.status_cache:
- # We don't want to block the request, so we could do this in background
- # but for now let's just do it.
- await self.update_statuses()
-
- results = []
- for iface in self.interfaces:
- status = self.status_cache.get(
- iface["name"],
- {"online": False, "last_check": 0},
- )
- results.append(
- {
- **iface,
- "online": status["online"],
- "last_check": status["last_check"],
- },
- )
-
- # Sort so online ones are first
- results.sort(key=lambda x: x["online"], reverse=True)
- return results
diff --git a/tests/backend/test_community_interfaces.py b/tests/backend/test_community_interfaces.py
index d79ce5ad..fb1f20ca 100644
--- a/tests/backend/test_community_interfaces.py
+++ b/tests/backend/test_community_interfaces.py
@@ -1,4 +1,4 @@
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock
import pytest
@@ -7,24 +7,14 @@ from meshchatx.src.backend.rnstatus_handler import RNStatusHandler
@pytest.mark.asyncio
-async def test_community_interfaces_manager_health_check():
+async def test_community_interfaces_manager_no_probe():
manager = CommunityInterfacesManager()
-
- # Mock check_health to return True for first, False for second
- with patch.object(
- CommunityInterfacesManager,
- "check_health",
- side_effect=[True, False],
- ):
- interfaces = await manager.get_interfaces()
-
- assert len(interfaces) == 2
- # First one should be online because we sort by online status
- assert interfaces[0]["online"] is True
- assert interfaces[1]["online"] is False
- # Check that we have both online and offline
- online_count = sum(1 for iface in interfaces if iface["online"])
- assert online_count == 1
+ interfaces = await manager.get_interfaces()
+ assert len(interfaces) >= 1
+ for iface in interfaces:
+ assert "name" in iface and "target_host" in iface and "target_port" in iface
+ assert iface.get("online") is None
+ assert iface.get("last_check") == 0
@pytest.mark.asyncio
@@ -59,18 +49,9 @@ async def test_rnstatus_integration_simulated():
@pytest.mark.asyncio
-async def test_community_interfaces_dynamic_update():
+async def test_community_interfaces_static_list():
manager = CommunityInterfacesManager()
-
- # Mock check_health to return different values over time
- with patch.object(CommunityInterfacesManager, "check_health") as mock_check:
- # First check: all online
- mock_check.return_value = True
- ifaces1 = await manager.get_interfaces()
- assert all(iface["online"] for iface in ifaces1)
-
- # Force update by clearing last_check and mock all offline
- manager.last_check = 0
- mock_check.return_value = False
- ifaces2 = await manager.get_interfaces()
- assert all(not iface["online"] for iface in ifaces2)
+ ifaces1 = await manager.get_interfaces()
+ ifaces2 = await manager.get_interfaces()
+ assert ifaces1 == ifaces2
+ assert all(iface.get("online") is None for iface in ifaces1)
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────